package net.wicp.tams.commons.binlog;

import java.io.IOException;
import java.io.InputStream;
import java.net.Socket;
import java.util.ArrayList;
import java.util.List;

import lombok.extern.slf4j.Slf4j;
import net.wicp.tams.commons.Result;
import net.wicp.tams.commons.apiext.SocketUtil;
import net.wicp.tams.commons.beans.Host;
import net.wicp.tams.commons.binlog.bean.PacketBean;
import net.wicp.tams.commons.binlog.bean.RsCol;
import net.wicp.tams.commons.binlog.constant.Command;
import net.wicp.tams.commons.binlog.constant.EventType;
import net.wicp.tams.commons.binlog.event.AbsEvent;
import net.wicp.tams.commons.binlog.impl.BinglogReadDump;
import net.wicp.tams.commons.binlog.replication.down.AbsResultPacket;
import net.wicp.tams.commons.binlog.replication.down.EOFPacket;
import net.wicp.tams.commons.binlog.replication.down.ErrorPacket;
import net.wicp.tams.commons.binlog.replication.down.GreetingPacket;
import net.wicp.tams.commons.binlog.replication.down.OKPacket;
import net.wicp.tams.commons.binlog.replication.down.RsFieldPacket;
import net.wicp.tams.commons.binlog.replication.down.RsHeaderPacket;
import net.wicp.tams.commons.binlog.replication.down.RsRowPacket;
import net.wicp.tams.commons.binlog.replication.up.SecurePasswordAuthenticationPacket;
import net.wicp.tams.commons.binlog.replication.up.command.AbsCommand;
import net.wicp.tams.commons.binlog.replication.up.command.ComBinlogDump;
import net.wicp.tams.commons.binlog.replication.up.command.ComInitDb;
import net.wicp.tams.commons.binlog.replication.up.command.ComQuery;
import net.wicp.tams.commons.binlog.replication.up.command.SingleCommand;
import net.wicp.tams.commons.io.OutputStreamBufferedWrite;

@Slf4j
public class Dump {
	private final String host;
	private final int port;
	private Socket socket;
	OutputStreamBufferedWrite out;
	InputStream in;
	private boolean isLogin = false;

	public Dump(String host, int port) {
		this.host = host;
		this.port = port;
		AbsEvent.hosts.set(Host.builder().hostIp(host).port(port).build());
	}

	public Dump(String host) {
		this(host, 3306);
	}

	public Result login(String user, String pwd, String defaultSchema) throws IOException {
		try {
			if (socket == null || socket.isClosed()) {
				socket = SocketUtil.create(host, port);
			}
			out = new OutputStreamBufferedWrite(socket.getOutputStream());
			in = socket.getInputStream();
		} catch (Exception e) {
			throw new IllegalAccessError(String.format("不能连接到主机:[%s]，端口：[%s]", host, port));
		}
		PacketBean greatBean = PacketBean.buildBean(in);
		if (greatBean.getPacketMarker() == ErrorPacket.PACKET_MARKER) {
			final ErrorPacket error = new ErrorPacket(greatBean);
			return Result.getError("欢迎包错误").setRetObjs(error);
		} else {
			GreetingPacket greetingPacket = new GreetingPacket(greatBean);
			SecurePasswordAuthenticationPacket pa = new SecurePasswordAuthenticationPacket(greetingPacket);
			pa.setUser(user);
			pa.setPassword(pwd);
			pa.setDefaultSchema(defaultSchema);
			byte[] bytes = pa.writeBody();
			out.write(bytes);
			out.flush();
			PacketBean loginRetBean = PacketBean.buildBean(in);
			if (loginRetBean.getPacketMarker() == ErrorPacket.PACKET_MARKER) {
				final ErrorPacket error = new ErrorPacket(loginRetBean);
				return Result.getError("登陆失败").setRetObjs(error);
			} else if (loginRetBean.getPacketMarker() == OKPacket.PACKET_MARKER) {
				isLogin = true;
				OKPacket oKPacket = new OKPacket(loginRetBean);
				return Result.getSuc().setRetObjs(oKPacket);
			} else {
				return Result.getError(String.format("不支持的MARKER：[%s]", loginRetBean.getPacketMarker()))
						.setRetObjs(loginRetBean);
			}
		}
	}

	/***
	 * 下载文件
	 * 
	 * @param binlogFilename
	 *            日志文件名
	 * @param binlogPos
	 *            日志文件位置
	 * @param ServerId
	 *            服务器ID
	 * @param curFile
	 *            是否只读当前文件 true:是，false：否
	 * @throws IOException
	 */
	@SuppressWarnings("resource")
	public void dumpLog(String binlogFilename, long binlogPos, long ServerId, boolean curFile) throws IOException {
		if (!isLogin) {
			throw new IllegalAccessError("没有登陆");
		}
		if (binlogPos < 4) {
			throw new IllegalAccessError("开始位置最小为4");
		}
		ComBinlogDump comBinlogDump = new ComBinlogDump();
		comBinlogDump.setBinlogFilename(binlogFilename);
		if (AbsEvent.hosts.get() == null || !EventType.isInit(AbsEvent.hosts.get())) {// 没有经过初始化则需要读一次FormatDescription头
			comBinlogDump.setBinlogPos(4);
		} else {
			comBinlogDump.setBinlogPos(binlogPos);
		}
		comBinlogDump.setFlags(0);
		comBinlogDump.setServerId(ServerId);
		byte[] input = comBinlogDump.writeBody();
		out.write(input);
		out.flush();
		PacketBean binlogPacketBean = PacketBean.buildEventBean(in, binlogFilename);
		if (binlogPacketBean.getPacketMarker() == ErrorPacket.PACKET_MARKER) {// 第一个是ROTATE_EVENT事件
			final ErrorPacket binPacket = new ErrorPacket(binlogPacketBean);
			log.error("下载binlog时出错：[{}]", binPacket.getBody().toString(true));
		} else {
			log.info("下载binlog成功，文件名：[{}],位置：[{}],服务ID:[{}],内容：[{}]", binlogFilename, binlogPos, ServerId,
					binlogPacketBean.getBody());
			// 解析binlog日志
			BinglogReadDump binlogParser = new BinglogReadDump(in);
			binlogParser.setFileName(binlogFilename);
			binlogParser.setCurFile(curFile);
			binlogParser.read(binlogPos);
			destory();
		}
	}

	public void dumpLog(String binlogFilename, long binlogPos, long ServerId) throws IOException {
		dumpLog(binlogFilename, binlogPos, ServerId, false);
	}

	public List<List<RsCol>> sqlQuery(String sql) throws IOException {
		if (!isLogin) {
			throw new IllegalAccessError("没有登陆");
		}
		ComQuery common = new ComQuery();
		common.setSql(sql);
		byte[] input = common.writeBody();
		out.write(input);
		out.flush();
		PacketBean packet = PacketBean.buildBean(in);
		List<List<RsCol>> retlist = new ArrayList<>();
		if (packet.getPacketMarker() == ErrorPacket.PACKET_MARKER) {
			final ErrorPacket binPacket = new ErrorPacket(packet);
			log.error("查询时出错：[{}]", binPacket.getBody().toString(true));
		} else {
			RsHeaderPacket rshead = new RsHeaderPacket(packet);
			System.out.println(rshead.getFieldCount());
			List<RsFieldPacket> fieldList = new ArrayList<>();
			while (true) {
				packet = PacketBean.buildBean(in);
				if (packet.getPacketMarker() == EOFPacket.PACKET_MARKER) {
					EOFPacket eof = new EOFPacket(packet);
					log.info("{}", eof);
					break;
				} else {
					RsFieldPacket field = new RsFieldPacket(packet);
					log.info("{}", field.getBody().toString(true));
					fieldList.add(field);
				}
			}

			while (true) {
				packet = PacketBean.buildBean(in);
				if (packet.getPacketMarker() == EOFPacket.PACKET_MARKER) {
					EOFPacket eof = new EOFPacket(packet);
					log.info("{}", eof);
					break;
				} else {
					List<RsCol> rowvalue = new ArrayList<>();
					RsRowPacket row = new RsRowPacket(packet);
					log.info("{}", row.getBody().toString(true));
					for (int i = 0; i < fieldList.size(); i++) {
						RsCol tempobj = RsCol.builder().columnType(fieldList.get(i).getFieldType())
								.valStr(row.getColumns().get(i)).build();
						rowvalue.add(tempobj);
					}
					retlist.add(rowvalue);
				}
			}
		}
		return retlist;
	}

	/***
	 * 发送单个命令
	 * 
	 * @param singleCommand
	 * @throws IOException
	 */
	public AbsResultPacket sendSingleCommand(Command command) throws IOException {
		if (!isLogin) {
			throw new IllegalAccessError("没有登陆");
		}
		SingleCommand singleCommand = new SingleCommand(command);
		sendCommon(singleCommand);
		return retResult();
	}

	public AbsResultPacket setDefaultDb(String defaultDb) throws IOException {
		if (!isLogin) {
			throw new IllegalAccessError("没有登陆");
		}
		ComInitDb c = new ComInitDb();
		c.setDefaultSchema(defaultDb);
		sendCommon(c);
		return retResult();
	}

	private void sendCommon(AbsCommand command) throws IOException {
		byte[] input = command.writeBody();
		out.write(input);
		out.flush();
	}

	private AbsResultPacket retResult() throws IOException {
		PacketBean packet = PacketBean.buildBean(in);
		if (packet.getPacketMarker() == ErrorPacket.PACKET_MARKER) {
			return new ErrorPacket(packet);
		} else if (packet.getPacketMarker() == OKPacket.PACKET_MARKER) {
			return new OKPacket(packet);
		} else {
			return null;
		}
	}

	public void destory() throws IOException {
		isLogin = false;
		if (in != null) {
			in.close();
		}
		if (out != null) {
			out.close();
		}
		if (socket != null) {
			socket.close();
		}
	}

}
